package com.nota.flink.cdc.main;


import com.nota.flink.cdc.common.config.MysqlCdcInfo;
import com.nota.flink.cdc.exception.BizException;
import com.nota.flink.cdc.pojo.BinlogInfo;
import com.nota.flink.cdc.pojo.DataChangeInfo;
import com.nota.flink.cdc.pojo.constants.RedisStreamConstants;
import com.nota.flink.cdc.utils.MysqlUtil;
import com.nota.flink.cdc.utils.RedisUtil;
import com.nota.flink.cdc.utils.SqlParserUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
@RequiredArgsConstructor
public class MysqlEventListener implements ApplicationRunner {

    private final DataChangeSink dataChangeSink;

    private final MysqlCdcInfo mysqlCdcInfo;

    @Resource
    private MysqlUtil mysqlUtil;

    @Value("${synchronously}")
    private Integer synchronously;

    /**
     * 关于 table的map  key:tableName ,value:主键
     */
    public Map<String, String> TABLE_MAP = new HashMap<>();

    public static AtomicInteger COUNT = new AtomicInteger(0);

    @Resource
    private RedisUtil redisUtil;


    /**
     * todo  手动调用
     *
     * @param args
     */
    @Override
    public void run(ApplicationArguments args) {
        CompletableFuture.runAsync(() -> {
            try {
                // 先初始化所有的表
                TABLE_MAP = initIndex();
                StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
                // 设置8个线程处理
                env.setParallelism(8);
                MySqlSource<DataChangeInfo> mySqlSource = buildDataChangeSource();
                DataStream<DataChangeInfo> streamSource = env
                        .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-source")
                        //对接收器使用并行1来保持消息的顺序
                        .setParallelism(1);
                streamSource.addSink(dataChangeSink);
                env.executeAsync("mysql-cdc-es");
                log.info("监控成功!!!!!!!!!!!");
            } catch (Exception e) {
                log.error("mysql --> es, Exception=", e);
            }
        }).exceptionally(ex -> {
            ex.printStackTrace();
            return null;
        });
    }


    /**
     * 构造变更数据源
     *
     * @return DebeziumSourceFunction<DataChangeInfo>
     */
    private MySqlSource<DataChangeInfo> buildDataChangeSource() {
        Properties properties = new Properties();
        properties.setProperty("converters", "dateConverters");
        properties.setProperty("dateConverters.type", "com.nota.flink.cdc.common.config.MySqlDateTimeConverter");
        //同步方式 1 全量且增量同步 2增量导入 3 按照时间戳导入
        if (synchronously == 1) {
            return getInitial(properties);
        } else if (synchronously == 2) {
            return getlatest(properties);
        } else if (synchronously == 3) {
            return getSpecificOffset(properties);
        } else {
            throw new BizException("配置有误");
        }


    }

    /**
     * 全量
     *
     * @return
     */
    private MySqlSource<DataChangeInfo> getInitial(Properties properties) {

        return MySqlSource.<DataChangeInfo>builder()
                .hostname(mysqlCdcInfo.getIp())
                .port(mysqlCdcInfo.getPort())
                .databaseList(mysqlCdcInfo.getDbs())
                // ddl语句也监控
                .includeSchemaChanges(true)
                //  新表也监控
                .scanNewlyAddedTableEnabled(true)
                // 支持正则匹配
                .tableList(mysqlCdcInfo.getTables())
                .username(mysqlCdcInfo.getUser())
                .password(mysqlCdcInfo.getPwd())

                /**
                 * initial初始化快照,即全量导入后增量导入(检测更新数据写入)
                 * latest:只进行增量导入(不读取历史变化)
                 * timestamp:指定时间戳进行数据导入(大于等于指定时间错读取数据)
                 */
                //指定位置模式
                //.startupOptions(StartupOptions.specificOffset("binlog.000003",197890))
                .startupOptions(StartupOptions.initial())
                .deserializer(new MysqlDeserialization())
                .debeziumProperties(properties)
                .serverTimeZone("Asia/Shanghai")
                .serverTimeZone("GMT+8")
                .build();
    }

    /**
     * 增量
     *
     * @return
     */
    private MySqlSource<DataChangeInfo> getlatest(Properties properties) {

        return MySqlSource.<DataChangeInfo>builder()
                .hostname(mysqlCdcInfo.getIp())
                .port(mysqlCdcInfo.getPort())
                .databaseList(mysqlCdcInfo.getDbs())
                // ddl语句也监控
                .includeSchemaChanges(true)
                //  新表也监控
                .scanNewlyAddedTableEnabled(true)
                // 支持正则匹配
                .tableList(mysqlCdcInfo.getTables())
                .username(mysqlCdcInfo.getUser())
                .password(mysqlCdcInfo.getPwd())

                /**
                 * initial初始化快照,即全量导入后增量导入(检测更新数据写入)
                 * latest:只进行增量导入(不读取历史变化)
                 * timestamp:指定时间戳进行数据导入(大于等于指定时间错读取数据)
                 */
                //指定位置模式
                //.startupOptions(StartupOptions.specificOffset("binlog.000003",197890))
                .startupOptions(StartupOptions.latest())
                .deserializer(new MysqlDeserialization())
                .debeziumProperties(properties)
                .serverTimeZone("Asia/Shanghai")
                .serverTimeZone("GMT+8")
                .build();
    }


    /**
     * 设置增量的
     *
     * @return
     */
    private MySqlSource<DataChangeInfo> getSpecificOffset(Properties properties) {
        Object binlogObj = redisUtil.getCacheObject(RedisStreamConstants.BINLOG);
        if (binlogObj == null) {
            //启动失败,需要先设置点参数
            throw new RuntimeException("按照位置启动失败,redis没有日志对象");
        }
        BinlogInfo dataChangeInfo = (BinlogInfo) binlogObj;
        return MySqlSource.<DataChangeInfo>builder()
                .hostname(mysqlCdcInfo.getIp())
                .port(mysqlCdcInfo.getPort())
                .databaseList(mysqlCdcInfo.getDbs())
                // ddl语句也监控
                .includeSchemaChanges(true)
                //  新表也监控
                .scanNewlyAddedTableEnabled(true)
                // 支持正则匹配
                .tableList(mysqlCdcInfo.getTables())
                .username(mysqlCdcInfo.getUser())
                .password(mysqlCdcInfo.getPwd())

                /**
                 * initial初始化快照,即全量导入后增量导入(检测更新数据写入)
                 * latest:只进行增量导入(不读取历史变化)
                 * timestamp:指定时间戳进行数据导入(大于等于指定时间错读取数据)
                 */
                //指定位置模式
                .startupOptions(StartupOptions.timestamp(dataChangeInfo.getOperatorTime()))
                .deserializer(new MysqlDeserialization())
                .debeziumProperties(properties)
                .serverTimeZone("Asia/Shanghai")
                .serverTimeZone("GMT+8")
                .build();


    }


    /**
     * 初始化所有的 index ,并且找到所有表主键设置到全局的map里
     */
    private Map<String, String> initIndex() {
        // 所有表
        List<String> tableList = mysqlUtil.getAllTable();
        Map<String, String> rtMap = new HashMap<>();
        // 获取所有表的 建表语句
        for (String tableName : tableList) {
            String ddl = mysqlUtil.showCreateTable(tableName);
            DataChangeInfo dataChangeInfo = new DataChangeInfo();
            dataChangeInfo.setDdl(ddl);
            try {
                SqlParserUtil.parseSql2(dataChangeInfo);
            } catch (Exception e) {
                log.warn("创建index[{}]失败", tableName, e.getMessage());
            }

            String primary = mysqlUtil.getPrimary(tableName);
            rtMap.put(tableName, primary);
        }
        return rtMap;
    }

    public Map<String, String> getMap() {
        return TABLE_MAP;
    }

}